package com.shujia.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Demo27PageRank {
  def main(args: Array[String]): Unit = {
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo27PageRank")
    conf.setMaster("local") // Spark 运行的方式

    // 构建Spark环境：SparkContext
    val sc: SparkContext = new SparkContext(conf)

    // 读取数据
    val pageRankRDD: RDD[String] = sc.textFile("spark/data/pageRank.txt")

    // 先对数据进行切分，将页面之间的关系 通过 一定的数据结构 保存
    val pageSourceRDD: RDD[(String, List[String])] = pageRankRDD.map(line => {
      val splits: Array[String] = line.split("->")
      val page: String = splits(0)
      (page, splits(1).split(",").toList)
    })

    var cnt: Int = 1
    val stop: Int = 10 // 控制循环次数
    // 给每个页面附上初始的PR值1
    var pageSourcePRRDD: RDD[(String, List[String], Double)] = pageSourceRDD.map(t2 => (t2._1, t2._2, 1.0))

    while (cnt <= stop) {
      // 将PR值平均分给每个出链指向的页面
      val pageRankNewPRRDD: RDD[(String, Double)] = pageSourcePRRDD.flatMap(t3 => {
        val avgPR: Double = t3._3 / t3._2.size
        t3._2.map(p => (p, avgPR))
      })
        // 统计每个页面新的PR值
        .reduceByKey(_ + _)

      // 关联pageSourceRDD 将新的PR值赋给每个页面 形成新的三元组
      val pageNewPRRDD: RDD[(String, List[String], Double)] = pageSourceRDD.join(pageRankNewPRRDD).map(t2 => (t2._1, t2._2._1, t2._2._2))
      pageSourcePRRDD = pageNewPRRDD
      println(s"经过${cnt}次循环之后，页面的PR值为")
      pageRankNewPRRDD.foreach(println)

      cnt += 1
    }


  }

}
